package com.example.klinemqflink.fun;

import com.example.klinemqflink.model.Kline;
import com.example.klinemqflink.model.MarketData;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

public class KlineWindowFunction extends ProcessWindowFunction<MarketData, Kline, String, TimeWindow> {

    private final String interval;

    public KlineWindowFunction(String interval) {
        this.interval = interval;
    }

    @Override
    public void process(String symbol, Context context,
                        Iterable<MarketData> marketData, Collector<Kline> out) {
        List<MarketData> dataPoints = new ArrayList<>();
        marketData.forEach(dataPoints::add);
        if (!dataPoints.isEmpty()) {
            MarketData first = dataPoints.get(0);
            MarketData last = dataPoints.get(dataPoints.size() - 1);

            double high = dataPoints.stream()
                    .max(Comparator.comparingDouble(MarketData::getPrice))
                    .get().getPrice();

            double low = dataPoints.stream()
                    .min(Comparator.comparingDouble(MarketData::getPrice))
                    .get().getPrice();

            double volume = dataPoints.stream()
                    .mapToDouble(MarketData::getVolume)
                    .sum();

            Kline kline = new Kline();
            out.collect(kline);
        }
    }
}
